package com.sdses.flink.apps;

import com.alibaba.fastjson2.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ververica.cdc.connectors.base.options.StartupOptions;
import com.ververica.cdc.connectors.base.source.jdbc.JdbcIncrementalSource;
import com.ververica.cdc.connectors.postgres.source.PostgresSourceBuilder;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.PreparedStatement;
import java.sql.SQLException;


public class PgCdc2Pg {
    public static final Logger logger = LoggerFactory.getLogger(PgCdc2Pg.class);

    public static void main(String[] args) throws Exception {

        ObjectMapper objectMapper = new ObjectMapper();

        JdbcIncrementalSource<DebeziumRecord> postgresIncrementalSource =
                PostgresSourceBuilder.PostgresIncrementalSource.<DebeziumRecord>builder()
                        .hostname("192.168.102.154")
                        .port(5866)
                        .database("db_datacube_sdses_test")
                        .schemaList("public")
                        .tableList("public.test_gis_erp_community2")
                        .username("sysdba")
                        .password("Sdses,hg2o23")
                        .slotName("flink")
                        .decodingPluginName("pgoutput")
                        .deserializer(new CustomDebeziumDeserializationSchema())
                        .includeSchemaChanges(true)
                        .startupOptions(StartupOptions.initial())
                        .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.enableCheckpointing(30000);

//        env.fromSource(postgresIncrementalSource, WatermarkStrategy.noWatermarks(), "PostgreSQLCDC")
//                .print("------------------------");


        env.fromSource(postgresIncrementalSource, WatermarkStrategy.noWatermarks(), "PostgreSQLCDC")
                .addSink(JdbcSink.sink(
                        "insert into public.test_gis_erp_community_sink (code, name, id, shape_length, shape_area, the_geom, import_time_) " +
                                "values (?,?,?,?,?,ST_GeomFromText(?),TO_TIMESTAMP(?, 'YYYY-MM-DD HH24:MI:SS.US')) " +
                                "ON CONFLICT (id) DO UPDATE SET " +
                                "name = EXCLUDED.name, " +
                                "code = EXCLUDED.code, " +
                                "shape_length = EXCLUDED.shape_length, " +
                                "shape_area = EXCLUDED.shape_area, " +
                                "the_geom = EXCLUDED.the_geom, " +
                                "import_time_ = EXCLUDED.import_time_",
                        new JdbcStatementBuilder<DebeziumRecord>() {
                            @Override
                            public void accept(PreparedStatement pStmt, DebeziumRecord record) throws SQLException {
                                JSONObject after = record.getAfter();
                                //处理insert/update操作
                                if (after!= null) {
                                    logger.info("PgCdc2Pg任务执行insert/update操作...");
                                    pStmt.setString(1, after.getString("code"));
                                    pStmt.setString(2, after.getString("name"));
                                    pStmt.setInt(3, after.getIntValue("id"));
                                    pStmt.setDouble(4, after.getDouble("shape_length"));
                                    pStmt.setDouble(5, after.getDouble("shape_area"));
                                    pStmt.setString(6, after.getString("the_geom"));
                                    pStmt.setString(7, after.getString("import_time_"));
//                                        String importTimeStr = after.getString("import_time_");
//                                        if (importTimeStr != null && !importTimeStr.isEmpty()) {
//                                            try {
//                                                java.sql.Timestamp timestamp = java.sql.Timestamp.valueOf(importTimeStr.substring(0, 19));
//                                                pStmt.setTimestamp(7, timestamp);
//                                            } catch (IllegalArgumentException e) {
//                                                logger.warn("无法将import_time_转换为Timestamp，字符串格式不正确: " + importTimeStr, e);
//                                                pStmt.setNull(7, Types.TIMESTAMP);
//                                            }
//                                        } else {
//                                            pStmt.setNull(7, Types.TIMESTAMP);
//                                        }

                                } else {
                                    //TODO 处理删除操作
                                    logger.info("PgCdc2Pg任务执行delete操作...");
                                    JSONObject before = record.getBefore();
                                    if (before!= null) {
                                        int idToDelete = before.getIntValue("id");
                                        // 创建一个新的PreparedStatement用于删除操作
                                        PreparedStatement deleteStmt = pStmt.getConnection().prepareStatement("DELETE FROM public.test_gis_erp_community_sink WHERE id =?");
                                        deleteStmt.setInt(1, idToDelete);
                                        deleteStmt.executeUpdate();
                                    }
                                }
                            }
                        },
                        JdbcExecutionOptions.builder()
                                .withBatchSize(10) // 批次大小,条数
                                .withBatchIntervalMs(5000) // 批次最大等待时间
                                .withMaxRetries(1) // 重复次数
                                .build(), // 写入参数配置,
                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                .withUrl("jdbc:postgresql://192.168.102.154:5866/db_datacube_sdses_test")
                                .withUsername("sysdba")
                                .withPassword("Sdses,hg2o23")
                                .withDriverName("org.postgresql.Driver")
                                .build()));

        env.execute();
    }
}
